package com.ericsson.enap.modules.procLog.service.impl;

import com.ericsson.common.domain.CollectBizTypeEnum;
import com.ericsson.common.mapper.JsonMapper;
import com.ericsson.common.pojo.FTPInfoVO;
import com.ericsson.common.util.*;
import com.ericsson.enap.modules.ossalarm.entity.OssAlarmMessage;
import com.ericsson.enap.modules.procLog.constant.ProcConstant;
import com.ericsson.enap.modules.procLog.entity.EdaAlarm;
import com.ericsson.enap.modules.procLog.entity.EnipAlarmPlatformConfig;
import com.ericsson.enap.modules.procLog.entity.EnipAlarmPlatformMonitor;
import com.ericsson.enap.modules.procLog.mapper.EnipAlarmPlatformConfigMapper;
import com.ericsson.enap.modules.procLog.mapper.EnipAlarmPlatformMonitorMapper;
import com.ericsson.enap.modules.procLog.mapper.ProcLogMapper;
import com.ericsson.enap.modules.procLog.service.ImsEdaAlarmService;
import com.ericsson.enap.modules.sys.entity.EnipNetelement;
import com.ericsson.enap.modules.sys.entity.EpcDict;
import com.ericsson.enap.modules.sys.service.EpcDictService;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.SftpATTRS;
import com.jcraft.jsch.SftpException;
import org.slf4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @description: EDA 告警功能service实现类
 * @author: yanghai
 * @create: 2021-07-29 09:52
 */
@Lazy(true)
@Service
public class ImsEdaAlarmServiceImpl implements ImsEdaAlarmService {

    private Logger log = LoggerUtils.Logger(LogFileName.ImsEdaAlarmJob);

    @Autowired
    private ProcLogMapper procLogMapper;
    @Autowired
    private EpcDictService epcDictService;
    @Autowired
    private EnipAlarmPlatformMonitorMapper enipAlarmPlatformMonitorMapper;
    @Autowired
    private EnipAlarmPlatformConfigMapper enipAlarmPlatformConfigMapper;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    //采集数据时间间隔
    private static long dataTime=5;

    @Override
    public void dealEdaAlarmFile(String collectorIp, String imsBossKpiPath, Date now, String delay) {
        //1.查询当前采集器配置了哪些网元需要进行下载
        log.info("======= bizType:"+11+" ======= collectorIp:"+collectorIp);
        List<EnipNetelement> list = procLogMapper.queryNetelementList(Integer.parseInt(CollectBizTypeEnum.IMSEDAAlarmDownLoad.value()),collectorIp);
        //2.逐个处理下载的网元
        FTPInfoVO fTPInfoVO;
        for(int i = 0;i<list.size();i++){
            fTPInfoVO = getFtpInfoVO(list, i);
            String localPath = imsBossKpiPath+ File.separator + list.get(i).getId() + File.separator;
            this.execute(Integer.parseInt(delay),now,list.get(i),localPath,fTPInfoVO);
        }
        log.info("===== All EDA Alarm kpi analysis files end! =====");
    }

    public void execute(int delay,Date date,EnipNetelement enipNetelement,String localPath,FTPInfoVO fTPInfoVO) {
        String beginTime = DateUtils.getStatisTime(delay,date,"yyyy-MM-dd_HH.mm");
        String endTime = DateUtils.getStatisTime(delay-5,date,"yyyy-MM-dd_HH.mm");
        String dateTime=DateUtils.getStatisTime(0,date,"yyyy-MM-dd HH:mm:ss");
        log.info("netId"+enipNetelement.getId()+"download start");
        //下载文件
        File edaAlarmFile=this.download(localPath,beginTime,endTime,fTPInfoVO,enipNetelement.getId());
        log.info("netId"+enipNetelement.getId()+"download end");
        Map<String,List<EdaAlarm>> resultMap=null;
        if(null==edaAlarmFile){
            this.log.info(enipNetelement.getId()+"null_"+"b:"+beginTime+",e:"+endTime);
        }else{
            log.info("netId"+enipNetelement.getId()+"analysisFile start");
            //解析文件

            resultMap=analysisFile(edaAlarmFile);
            log.info("netId"+enipNetelement.getId()+"analysisFile end:"+resultMap);
        }
        //读取EDA告警的配置信息
        EnipAlarmPlatformConfig enipAlarmPlatformConfig=new EnipAlarmPlatformConfig();
        enipAlarmPlatformConfig.setAlarmBusinessType(1l);
        enipAlarmPlatformConfig.setEnable(1);
        List<EnipAlarmPlatformConfig> enipAlarmPlatformConfigs=enipAlarmPlatformConfigMapper.selectEnipAlarmPlatformConfigList(enipAlarmPlatformConfig);
        if(resultMap!=null&&resultMap.size()>0&&enipAlarmPlatformConfigs.size()>0){
            log.info("netId"+enipNetelement.getId()+"analysis Alarm start");
            dealAlarm(enipAlarmPlatformConfigs,resultMap,enipNetelement,dateTime);
            log.info("netId"+enipNetelement.getId()+"analysis Alarm end");
        }else{
            log.error("读取文件数据失败或告警配置不存在");
        }
    }

    /**
     * 根据数据处理告警信息
     */
    private void dealAlarm(List<EnipAlarmPlatformConfig> enipAlarmPlatformConfigs,Map<String,List<EdaAlarm>> resultMap,EnipNetelement enipNetelement,String dateTime){
        Map<Long,EnipAlarmPlatformConfig> enipAlarmPlatformConfigMap=enipAlarmPlatformConfigs.stream().collect(Collectors.toMap(EnipAlarmPlatformConfig::getTemplateId, a -> a));
        for (Map.Entry<String, List<EdaAlarm>> entry : resultMap.entrySet()) {
            String alarmType=entry.getKey();
            List<EdaAlarm> alarmValue=entry.getValue();
            switch(alarmType){
                case "northEvent":
                    if(enipAlarmPlatformConfigMap.containsKey(4l)) {//重要指令执行成功率
                        dealNorthEventAlarm(enipAlarmPlatformConfigMap, alarmValue,enipNetelement,dateTime);
                    }
                    break;
                case "northPl":
                    //EDA板卡指令成功率低于门限:1、EDA板卡指令执行时间高于门限:2
                     if(enipAlarmPlatformConfigMap.containsKey(1l)||enipAlarmPlatformConfigMap.containsKey(2l)) {
                        dealNorthPlAlarm(enipAlarmPlatformConfigMap, alarmValue,enipNetelement,dateTime);
                    }
                    break;
                case "northErrorCode":
                    //EDA北向指令异常错误码高于门限
                    if(enipAlarmPlatformConfigMap.containsKey(3l)) {
                        dealNorthErrorCodeAlarm(enipAlarmPlatformConfigMap, alarmValue,enipNetelement,dateTime);
                    }
                    break;
            }
        }
    }

    /**
     * 处理北向事件告警（联指重要指令执行成功率低）
     * @param enipAlarmPlatformConfigMap
     * @param alarmValue
     */
    private void dealNorthEventAlarm(Map<Long,EnipAlarmPlatformConfig> enipAlarmPlatformConfigMap,List<EdaAlarm> alarmValue,EnipNetelement enipNetelement,String dateTime){
        EnipAlarmPlatformConfig enipAlarmPlatformConfig=enipAlarmPlatformConfigMap.get(4l);
        EnipAlarmPlatformMonitor enipAlarmPlatformMonitor=new EnipAlarmPlatformMonitor();
        enipAlarmPlatformMonitor.setNetId(enipNetelement.getId());
        enipAlarmPlatformMonitor.setConfigId(enipAlarmPlatformConfig.getId());
        enipAlarmPlatformMonitor.setStatus(1);
        List<EnipAlarmPlatformMonitor> enipAlarmPlatformMonitors1=enipAlarmPlatformMonitorMapper.selectEnipAlarmPlatformMonitorList(enipAlarmPlatformMonitor);
        enipAlarmPlatformMonitor.setStatus(2);
        List<EnipAlarmPlatformMonitor> enipAlarmPlatformMonitors2=enipAlarmPlatformMonitorMapper.selectEnipAlarmPlatformMonitorList(enipAlarmPlatformMonitor);
        boolean checkResult=true;
        String []eventStr=enipAlarmPlatformConfig.getOtherCondition2().split("\\|");
        Set set = new HashSet(Arrays.asList(eventStr));
        EnipAlarmPlatformMonitor enipAlarmPlatformMonitorData;
        if (enipAlarmPlatformMonitors1.size()>0){
            //该网元该告警类型已经发送告警平台,
            // 与配置表的clean_threshold，做比较判断是否满足清楚告警的条件，
            // 判断规则：每条数据指令（ADD_SBR、MOD_SBR、SET_OWSBR、ACT_BAIC、DEA_BAIC、ACT_BAOC、DEA_BAOC(other_condition2)中）告警数量大于 20条（other_condition1）数  成功率高于85%(clean_threshold)    单条成功
            //每条数据指令告警数量小于 20条（other_condition1）数    单条成功
            //按上判断都成功,做如下清除判断
            // (1)将数据时间间隔静态变量(dataTime)与数据库中数据表的clean_minute字段做和与配置表clean_minute清楚门限作比较，
            // (2)如果>=配置表clean_minute，向告警平台发送清楚告警信息 status=0，否则将计算结果赋予数据表的clean_minute status=1
            //如果不都成功，将数据表中的clean_minute字段置0
            enipAlarmPlatformMonitorData=enipAlarmPlatformMonitors1.get(0);
            for(EdaAlarm edaAlarm:alarmValue){
                if(set.contains(edaAlarm.getEventName())){
                    if((Double.parseDouble(edaAlarm.getTotalCount())>=Double.parseDouble(enipAlarmPlatformConfig.getOtherCondition1())&&
                            Double.parseDouble(edaAlarm.getRateSuc())>=Double.parseDouble(enipAlarmPlatformConfig.getCleanThreshold()))||
                            Double.parseDouble(edaAlarm.getTotalCount())<=Double.parseDouble(enipAlarmPlatformConfig.getOtherCondition1())){
                        checkResult=true;
                    }else{
                        checkResult=false;
                        break;
                    }
                }
            }
            if(checkResult){
                long timeTemp=enipAlarmPlatformMonitorData.getCleanMinute()+dataTime;
                enipAlarmPlatformMonitorData.setCleanMinute(timeTemp);
                if(timeTemp>=enipAlarmPlatformConfig.getCleanMinute()){
                    enipAlarmPlatformMonitorData.setStatus(0);
                    //发送kafka清除告警
                    String sendMsg=createMqMessage(enipAlarmPlatformConfig,enipAlarmPlatformMonitorData,enipNetelement);
                    enipAlarmPlatformMonitorData.setCancelAlarmMsg(sendMsg);
                    sendAlarm(sendMsg);
                }
            }else{
                enipAlarmPlatformMonitorData.setCleanMinute(0l);
            }
            enipAlarmPlatformMonitorData.setUpdateTime(dateTime);
            enipAlarmPlatformMonitorMapper.updateEnipAlarmPlatformMonitor(enipAlarmPlatformMonitorData);
            log.info("dealNorthEventAlarm list1 update:"+enipAlarmPlatformMonitorData.toString());
        }else if(enipAlarmPlatformMonitors2.size()>0){
            //该网元该告警类型未发送告警平台，
            //与配置表的appear_threshold，做比较判断是否为告警数据
            //判断规则：ADD_SBR、MOD_SBR、SET_OWSBR、ACT_BAIC、DEA_BAIC、ACT_BAOC、DEA_BAOC(other_condition2)中任意一类指令数量超过20（other_condition1）条，且成功率低于85%(appear_threshold)判断为异常
            //1、如果不是告警数据：将数据库中该条数据删除
            //2、如果是告警数据，执行以下操作
            //将数据时间间隔静态变量(dataTime)与数据库中数据表的appear_minute字段做和与配置表appear_minute告警门限作比较，
            //如果>=配置表appear_minute，向告警平台发送告警信息 status=1，否则将计算结果赋予数据表的appear_minute, status=2
            enipAlarmPlatformMonitorData=enipAlarmPlatformMonitors2.get(0);
            for(EdaAlarm edaAlarm:alarmValue){
                if(set.contains(edaAlarm.getEventName())){
                    if(Double.parseDouble(edaAlarm.getTotalCount())>=Double.parseDouble(enipAlarmPlatformConfig.getOtherCondition1())&&
                            Double.parseDouble(edaAlarm.getRateSuc())<=Double.parseDouble(enipAlarmPlatformConfig.getAppearThreshold())){
                        checkResult=false;
                        break;
                    }
                }
            }
            if(checkResult){
                enipAlarmPlatformMonitorMapper.deleteEnipAlarmPlatformMonitorById(enipAlarmPlatformMonitorData.getId());
                log.info("dealNorthEventAlarm list2 delete:"+enipAlarmPlatformMonitorData.toString());
            }else{
                long timeTemp=enipAlarmPlatformMonitorData.getAppearMinute()+dataTime;
                enipAlarmPlatformMonitorData.setAppearMinute(timeTemp);
                if(timeTemp>=enipAlarmPlatformConfig.getAppearMinute()){
                    enipAlarmPlatformMonitorData.setStatus(1);
                    //发送kafka产生告警
                    String sendMsg=createMqMessage(enipAlarmPlatformConfig,enipAlarmPlatformMonitorData,enipNetelement);
                    enipAlarmPlatformMonitorData.setCreateAlarmMsg(sendMsg);
                    sendAlarm(sendMsg);
                }
                enipAlarmPlatformMonitorData.setUpdateTime(dateTime);
                enipAlarmPlatformMonitorMapper.updateEnipAlarmPlatformMonitor(enipAlarmPlatformMonitorData);
                log.info("dealNorthEventAlarm list2 update:"+enipAlarmPlatformMonitorData.toString());
            }
        }else{
            //该网元该告警类型数据库中不存在
            //与配置表的appear_threshold，做比较判断是否为告警数据
            //判断规则：ADD_SBR、MOD_SBR、SET_OWSBR、ACT_BAIC、DEA_BAIC、ACT_BAOC、DEA_BAOC(other_condition2)中任意一类指令数量超过20（other_condition1）条，且成功率低于85%(appear_threshold)判断为异常
            //2、如果是告警数据，执行以下操作
            //将数据时间间隔静态变量(dataTime)与配置表appear_minute告警门限作比较，
            //如果>=配置表appear_minute，向告警平台发送告警信息并存入数据库 status=1，否则将数据存入数据表中 status=2
            enipAlarmPlatformMonitorData=new EnipAlarmPlatformMonitor();
            for(EdaAlarm edaAlarm:alarmValue){
                if(set.contains(edaAlarm.getEventName())){
                    if(Integer.parseInt(edaAlarm.getTotalCount())>=Integer.parseInt(enipAlarmPlatformConfig.getOtherCondition1())&&
                            Double.parseDouble(edaAlarm.getRateSuc())<=Double.parseDouble(enipAlarmPlatformConfig.getAppearThreshold())){
                        checkResult=false;
                        break;
                    }
                }
            }
            if(!checkResult){
                enipAlarmPlatformMonitorData.setConfigId(enipAlarmPlatformConfig.getId());
                enipAlarmPlatformMonitorData.setNetId(enipNetelement.getId());
                enipAlarmPlatformMonitorData.setAlarmId("EDA_"+Jdk8DateUtils.now().formatDateTime("yyyyMMddHHmmssSSS"));
                enipAlarmPlatformMonitorData.setCreateTime(dateTime);
                enipAlarmPlatformMonitorData.setCleanMinute(0l);
                long timeTemp=dataTime;
                enipAlarmPlatformMonitorData.setAppearMinute(timeTemp);
                if(timeTemp>=enipAlarmPlatformConfig.getAppearMinute()){
                    enipAlarmPlatformMonitorData.setStatus(1);
                    //发送kafka产生告警
                    String sendMsg=createMqMessage(enipAlarmPlatformConfig,enipAlarmPlatformMonitorData,enipNetelement);
                    enipAlarmPlatformMonitorData.setCreateAlarmMsg(sendMsg);
                    sendAlarm(sendMsg);
                }else{
                    enipAlarmPlatformMonitorData.setStatus(2);
                }
                enipAlarmPlatformMonitorMapper.insertEnipAlarmPlatformMonitor(enipAlarmPlatformMonitorData);
                log.info("dealNorthEventAlarm insert :"+enipAlarmPlatformMonitorData.toString());
            }
        }
    }

    /**
     * 处理北向板卡告警（PL板卡联指执行成功率低、联指执行时间过长）
     * @param enipAlarmPlatformConfigMap
     * @param alarmValue
     */
    private void dealNorthPlAlarm(Map<Long,EnipAlarmPlatformConfig> enipAlarmPlatformConfigMap,List<EdaAlarm> alarmValue,EnipNetelement enipNetelement,String dateTime){
        for (Map.Entry<Long, EnipAlarmPlatformConfig> entry : enipAlarmPlatformConfigMap.entrySet()) {
            Long alarmType=entry.getKey();
            EnipAlarmPlatformConfig enipAlarmPlatformConfig=entry.getValue();
            if(alarmType.intValue()==1){//板卡指令成功率
                dealNorthPlRateAlarm(enipAlarmPlatformConfig,alarmValue,enipNetelement,dateTime);
            }else if(alarmType.intValue()==2){//板卡指令执行时间
                dealNorthPlTimeAlarm(enipAlarmPlatformConfig,alarmValue,enipNetelement,dateTime);
            }
        }
    }

    /**
     * 处理板卡成功率
     * 单个PL板卡5分钟粒度的指令数量超过20(other_condition1)条，且成功率低于60%(appear_threshold)，判断为异常。
     * @param enipAlarmPlatformConfig
     * @param alarmValue
     * @param enipNetelement
     * @param dateTime
     */
    private void dealNorthPlRateAlarm(EnipAlarmPlatformConfig enipAlarmPlatformConfig,List<EdaAlarm> alarmValue,EnipNetelement enipNetelement,String dateTime){
        for(EdaAlarm edaAlarm:alarmValue){
            EnipAlarmPlatformMonitor enipAlarmPlatformMonitor=new EnipAlarmPlatformMonitor();
            enipAlarmPlatformMonitor.setNetId(enipNetelement.getId());
            enipAlarmPlatformMonitor.setMemo(edaAlarm.getPl());//板卡
            enipAlarmPlatformMonitor.setConfigId(enipAlarmPlatformConfig.getId());
            enipAlarmPlatformMonitor.setStatus(1);
            List<EnipAlarmPlatformMonitor> enipAlarmPlatformMonitors1=enipAlarmPlatformMonitorMapper.selectEnipAlarmPlatformMonitorList(enipAlarmPlatformMonitor);
            enipAlarmPlatformMonitor.setStatus(2);
            List<EnipAlarmPlatformMonitor> enipAlarmPlatformMonitors2=enipAlarmPlatformMonitorMapper.selectEnipAlarmPlatformMonitorList(enipAlarmPlatformMonitor);
            EnipAlarmPlatformMonitor enipAlarmPlatformMonitorData;
            if (enipAlarmPlatformMonitors1.size()>0){
                enipAlarmPlatformMonitorData=enipAlarmPlatformMonitors1.get(0);
                if((Long.parseLong(edaAlarm.getTotalCount())>=Long.parseLong(enipAlarmPlatformConfig.getOtherCondition1())&&
                        Double.parseDouble(edaAlarm.getRateSuc())>=Double.parseDouble(enipAlarmPlatformConfig.getCleanThreshold()))||
                        Long.parseLong(edaAlarm.getTotalCount())<=Long.parseLong(enipAlarmPlatformConfig.getOtherCondition1())){
                    long timeTemp=enipAlarmPlatformMonitorData.getCleanMinute()+dataTime;
                    enipAlarmPlatformMonitorData.setCleanMinute(timeTemp);
                    if(timeTemp>=enipAlarmPlatformConfig.getCleanMinute()){
                        enipAlarmPlatformMonitorData.setStatus(0);
                        //发送kafka清除告警
                        String sendMsg=createMqMessage(enipAlarmPlatformConfig,enipAlarmPlatformMonitorData,enipNetelement);
                        enipAlarmPlatformMonitorData.setCancelAlarmMsg(sendMsg);
                        sendAlarm(sendMsg);
                    }
                }else{
                    enipAlarmPlatformMonitorData.setCleanMinute(0l);
                }
                enipAlarmPlatformMonitorData.setUpdateTime(dateTime);
                enipAlarmPlatformMonitorMapper.updateEnipAlarmPlatformMonitor(enipAlarmPlatformMonitorData);
                log.info("dealNorthPlRateAlarm list1 update:"+enipAlarmPlatformMonitorData.toString());
            }else if(enipAlarmPlatformMonitors2.size()>0){
                enipAlarmPlatformMonitorData=enipAlarmPlatformMonitors2.get(0);
                if((Long.parseLong(edaAlarm.getTotalCount())>=Long.parseLong(enipAlarmPlatformConfig.getOtherCondition1())&&
                        Double.parseDouble(edaAlarm.getRateSuc())<=Double.parseDouble(enipAlarmPlatformConfig.getAppearThreshold()))){
                    long timeTemp=enipAlarmPlatformMonitorData.getAppearMinute()+dataTime;
                    enipAlarmPlatformMonitorData.setAppearMinute(timeTemp);
                    if(timeTemp>=enipAlarmPlatformConfig.getAppearMinute()){
                        enipAlarmPlatformMonitorData.setStatus(1);
                        //发送kafka产生告警
                        String sendMsg=createMqMessage(enipAlarmPlatformConfig,enipAlarmPlatformMonitorData,enipNetelement);
                        enipAlarmPlatformMonitorData.setCreateAlarmMsg(sendMsg);
                        sendAlarm(sendMsg);
                    }
                    enipAlarmPlatformMonitorData.setUpdateTime(dateTime);
                    enipAlarmPlatformMonitorMapper.updateEnipAlarmPlatformMonitor(enipAlarmPlatformMonitorData);
                    log.info("dealNorthPlRateAlarm list2 update:"+enipAlarmPlatformMonitorData.toString());
                }else{
                    enipAlarmPlatformMonitorMapper.deleteEnipAlarmPlatformMonitorById(enipAlarmPlatformMonitorData.getId());
                    log.info("dealNorthPlRateAlarm list2 delete:"+enipAlarmPlatformMonitorData.toString());
                }
            }else{
                if((Double.parseDouble(edaAlarm.getTotalCount())>=Double.parseDouble(enipAlarmPlatformConfig.getOtherCondition1())&&
                        Double.parseDouble(edaAlarm.getRateSuc())<=Double.parseDouble(enipAlarmPlatformConfig.getAppearThreshold()))) {
                    enipAlarmPlatformMonitorData = new EnipAlarmPlatformMonitor();
                    enipAlarmPlatformMonitorData.setConfigId(enipAlarmPlatformConfig.getId());
                    enipAlarmPlatformMonitorData.setNetId(enipNetelement.getId());
                    enipAlarmPlatformMonitorData.setMemo(edaAlarm.getPl());
                    enipAlarmPlatformMonitorData.setAlarmId("EDA_"+Jdk8DateUtils.now().formatDateTime("yyyyMMddHHmmssSSS"));
                    enipAlarmPlatformMonitorData.setCreateTime(dateTime);
                    long timeTemp=dataTime;
                    enipAlarmPlatformMonitorData.setCleanMinute(0l);
                    enipAlarmPlatformMonitorData.setAppearMinute(timeTemp);
                    if(timeTemp>=enipAlarmPlatformConfig.getAppearMinute()){
                        enipAlarmPlatformMonitorData.setStatus(1);
                        //发送kafka产生告警
                        String sendMsg=createMqMessage(enipAlarmPlatformConfig,enipAlarmPlatformMonitorData,enipNetelement);
                        enipAlarmPlatformMonitorData.setCreateAlarmMsg(sendMsg);
                        sendAlarm(sendMsg);
                    }else{
                        enipAlarmPlatformMonitorData.setStatus(2);
                    }
                    enipAlarmPlatformMonitorMapper.insertEnipAlarmPlatformMonitor(enipAlarmPlatformMonitorData);
                    log.info("dealNorthPlRateAlarm insert:"+enipAlarmPlatformMonitorData.toString());
                }
            }
        }
    }

    /**
     * 处理板卡执行时间告警
     * 每个PL板卡5分钟粒度内每条指令平均执行时长超过1.5(AppearThreshold)秒，判断为异常。
     * @param enipAlarmPlatformConfig
     * @param alarmValue
     * @param enipNetelement
     * @param dateTime
     */
    private void dealNorthPlTimeAlarm(EnipAlarmPlatformConfig enipAlarmPlatformConfig,List<EdaAlarm> alarmValue,EnipNetelement enipNetelement,String dateTime){
        for(EdaAlarm edaAlarm:alarmValue){
            EnipAlarmPlatformMonitor enipAlarmPlatformMonitor=new EnipAlarmPlatformMonitor();
            enipAlarmPlatformMonitor.setNetId(enipNetelement.getId());
            enipAlarmPlatformMonitor.setMemo(edaAlarm.getPl());//板卡
            enipAlarmPlatformMonitor.setConfigId(enipAlarmPlatformConfig.getId());
            enipAlarmPlatformMonitor.setStatus(1);
            List<EnipAlarmPlatformMonitor> enipAlarmPlatformMonitors1=enipAlarmPlatformMonitorMapper.selectEnipAlarmPlatformMonitorList(enipAlarmPlatformMonitor);
            enipAlarmPlatformMonitor.setStatus(2);
            List<EnipAlarmPlatformMonitor> enipAlarmPlatformMonitors2=enipAlarmPlatformMonitorMapper.selectEnipAlarmPlatformMonitorList(enipAlarmPlatformMonitor);
            EnipAlarmPlatformMonitor enipAlarmPlatformMonitorData;
            double commandRunTime=Double.parseDouble(edaAlarm.getTimeTotal())/Double.parseDouble(edaAlarm.getTotalCount());
            if (enipAlarmPlatformMonitors1.size()>0){
                enipAlarmPlatformMonitorData=enipAlarmPlatformMonitors1.get(0);
                if(commandRunTime<Double.parseDouble(enipAlarmPlatformConfig.getCleanThreshold())){
                    long timeTemp=enipAlarmPlatformMonitorData.getCleanMinute()+dataTime;
                    enipAlarmPlatformMonitorData.setCleanMinute(timeTemp);
                    if(timeTemp>=enipAlarmPlatformConfig.getCleanMinute()){
                        enipAlarmPlatformMonitorData.setStatus(0);
                        //发送kafka清除告警
                        String sendMsg=createMqMessage(enipAlarmPlatformConfig,enipAlarmPlatformMonitorData,enipNetelement);
                        enipAlarmPlatformMonitorData.setCancelAlarmMsg(sendMsg);
                        sendAlarm(sendMsg);
                    }
                }else{
                    enipAlarmPlatformMonitorData.setCleanMinute(0l);
                }
                enipAlarmPlatformMonitorData.setUpdateTime(dateTime);
                enipAlarmPlatformMonitorMapper.updateEnipAlarmPlatformMonitor(enipAlarmPlatformMonitorData);
                log.info("dealNorthPlTimeAlarm list1 update:"+enipAlarmPlatformMonitorData.toString());
            }else if(enipAlarmPlatformMonitors2.size()>0){
                enipAlarmPlatformMonitorData=enipAlarmPlatformMonitors2.get(0);
                if(commandRunTime>Double.parseDouble(enipAlarmPlatformConfig.getAppearThreshold())){
                    long timeTemp=enipAlarmPlatformMonitorData.getAppearMinute()+dataTime;
                    enipAlarmPlatformMonitorData.setAppearMinute(timeTemp);
                    if(timeTemp>=enipAlarmPlatformConfig.getAppearMinute()){
                        enipAlarmPlatformMonitorData.setStatus(1);
                        //发送kafka产生告警
                        String sendMsg=createMqMessage(enipAlarmPlatformConfig,enipAlarmPlatformMonitorData,enipNetelement);
                        enipAlarmPlatformMonitorData.setCreateAlarmMsg(sendMsg);
                        sendAlarm(sendMsg);
                    }
                    enipAlarmPlatformMonitorData.setUpdateTime(dateTime);
                    enipAlarmPlatformMonitorMapper.updateEnipAlarmPlatformMonitor(enipAlarmPlatformMonitorData);
                    log.info("dealNorthPlTimeAlarm list2 update:"+enipAlarmPlatformMonitorData.toString());
                }else{
                    enipAlarmPlatformMonitorMapper.deleteEnipAlarmPlatformMonitorById(enipAlarmPlatformMonitorData.getId());
                    log.info("dealNorthPlTimeAlarm list2 delete:"+enipAlarmPlatformMonitorData.toString());
                }
            }else{
                if(commandRunTime>=Double.parseDouble(enipAlarmPlatformConfig.getAppearThreshold())) {
                    enipAlarmPlatformMonitorData = new EnipAlarmPlatformMonitor();
                    enipAlarmPlatformMonitorData.setConfigId(enipAlarmPlatformConfig.getId());
                    enipAlarmPlatformMonitorData.setNetId(enipNetelement.getId());
                    enipAlarmPlatformMonitorData.setMemo(edaAlarm.getPl());
                    enipAlarmPlatformMonitorData.setAlarmId("EDA_"+Jdk8DateUtils.now().formatDateTime("yyyyMMddHHmmssSSS"));
                    enipAlarmPlatformMonitorData.setCreateTime(dateTime);
                    enipAlarmPlatformMonitorData.setCleanMinute(0l);
                    long timeTemp=dataTime;
                    enipAlarmPlatformMonitorData.setAppearMinute(timeTemp);
                    if(timeTemp>=enipAlarmPlatformConfig.getAppearMinute()){
                        enipAlarmPlatformMonitorData.setStatus(1);
                        //发送kafka产生告警
                        String sendMsg=createMqMessage(enipAlarmPlatformConfig,enipAlarmPlatformMonitorData,enipNetelement);
                        enipAlarmPlatformMonitorData.setCreateAlarmMsg(sendMsg);
                        sendAlarm(sendMsg);
                    }else{
                        enipAlarmPlatformMonitorData.setStatus(2);
                    }
                    enipAlarmPlatformMonitorMapper.insertEnipAlarmPlatformMonitor(enipAlarmPlatformMonitorData);
                    log.info("dealNorthPlTimeAlarm insert:"+enipAlarmPlatformMonitorData.toString());
                }
            }
        }
    }

    /**
     * 处理错误码告警（联指出现异常报错）
     * @param enipAlarmPlatformConfigMap
     * @param alarmValue
     */
    private void dealNorthErrorCodeAlarm(Map<Long,EnipAlarmPlatformConfig> enipAlarmPlatformConfigMap,List<EdaAlarm> alarmValue,EnipNetelement enipNetelement,String dateTime){
        EnipAlarmPlatformConfig enipAlarmPlatformConfig=enipAlarmPlatformConfigMap.get(3l);
        EnipAlarmPlatformMonitor enipAlarmPlatformMonitor=new EnipAlarmPlatformMonitor();
        enipAlarmPlatformMonitor.setNetId(enipNetelement.getId());
        enipAlarmPlatformMonitor.setConfigId(enipAlarmPlatformConfig.getId());
        enipAlarmPlatformMonitor.setStatus(1);
        List<EnipAlarmPlatformMonitor> enipAlarmPlatformMonitors1=enipAlarmPlatformMonitorMapper.selectEnipAlarmPlatformMonitorList(enipAlarmPlatformMonitor);
        enipAlarmPlatformMonitor.setStatus(2);
        List<EnipAlarmPlatformMonitor> enipAlarmPlatformMonitors2=enipAlarmPlatformMonitorMapper.selectEnipAlarmPlatformMonitorList(enipAlarmPlatformMonitor);
        String []errorCode=enipAlarmPlatformConfig.getOtherCondition1().split("\\|");
        Set set = new HashSet(Arrays.asList(errorCode));
        EnipAlarmPlatformMonitor enipAlarmPlatformMonitorData;
        if (enipAlarmPlatformMonitors1.size()>0){
            //该网元该告警类型已经发送告警平台,
            // 与配置表的clean_threshold，做比较判断是否满足清楚告警的条件，
            // 判断规则：排除掉数据中other_condition1中的正常错误码，异常错误码数量小于clean_threshold
            //按上判断都成功,做如下清除判断
            // (1)将数据时间间隔静态变量(dataTime)与数据库中数据表的clean_minute字段做和与配置表clean_minute清楚门限作比较，
            // (2)如果>=配置表clean_minute，向告警平台发送清楚告警信息 status=0，否则将计算结果赋予数据表的clean_minute status=1
            //如果不都成功，将数据表中的clean_minute字段置0
            enipAlarmPlatformMonitorData=enipAlarmPlatformMonitors1.get(0);
            double tempTotalCount=0;
            for(EdaAlarm edaAlarm:alarmValue){
                if(!set.contains(edaAlarm.getCode())){
                    tempTotalCount=tempTotalCount+Double.parseDouble(edaAlarm.getTotalCount());
                }
            }
            if(tempTotalCount<=Double.parseDouble(enipAlarmPlatformConfig.getCleanThreshold())){
                long timeTemp=enipAlarmPlatformMonitorData.getCleanMinute()+dataTime;
                enipAlarmPlatformMonitorData.setCleanMinute(timeTemp);
                if(timeTemp>=enipAlarmPlatformConfig.getCleanMinute()){
                    enipAlarmPlatformMonitorData.setStatus(0);
                    //发送kafka清除告警
                    String sendMsg=createMqMessage(enipAlarmPlatformConfig,enipAlarmPlatformMonitorData,enipNetelement);
                    enipAlarmPlatformMonitorData.setCancelAlarmMsg(sendMsg);
                    sendAlarm(sendMsg);
                }
            }else{
                enipAlarmPlatformMonitorData.setCleanMinute(0l);
            }
            enipAlarmPlatformMonitorData.setUpdateTime(dateTime);
            enipAlarmPlatformMonitorMapper.updateEnipAlarmPlatformMonitor(enipAlarmPlatformMonitorData);
            log.info("dealNorthErrorCodeAlarm list1 update:"+enipAlarmPlatformMonitorData.toString());
        }else if(enipAlarmPlatformMonitors2.size()>0){
            //该网元该告警类型未发送告警平台，
            //与配置表的appear_threshold，做比较判断是否为告警数据
            //判断规则：排除掉数据中other_condition1中的正常错误码，异常错误码数量大于appear_threshold
            //1、如果不是告警数据：将数据库中该条数据删除
            //2、如果是告警数据，执行以下操作
            //将数据时间间隔静态变量(dataTime)与数据库中数据表的appear_minute字段做和与配置表appear_minute告警门限作比较，
            //如果>=配置表appear_minute，向告警平台发送告警信息 status=1，否则将计算结果赋予数据表的appear_minute, status=2
            enipAlarmPlatformMonitorData=enipAlarmPlatformMonitors2.get(0);
            long tempTotalCount=0;
            for(EdaAlarm edaAlarm:alarmValue){
                if(!set.contains(edaAlarm.getCode())){
                    tempTotalCount=tempTotalCount+Long.parseLong(edaAlarm.getTotalCount());
                }
            }
            if(tempTotalCount<=Double.parseDouble(enipAlarmPlatformConfig.getAppearThreshold())){
                enipAlarmPlatformMonitorMapper.deleteEnipAlarmPlatformMonitorById(enipAlarmPlatformMonitorData.getId());
                log.info("dealNorthErrorCodeAlarm list2 delete:"+enipAlarmPlatformMonitorData.toString());
            }else{
                long timeTemp=enipAlarmPlatformMonitorData.getAppearMinute()+dataTime;
                enipAlarmPlatformMonitorData.setAppearMinute(timeTemp);
                if(timeTemp>=enipAlarmPlatformConfig.getAppearMinute()){
                    enipAlarmPlatformMonitorData.setStatus(1);
                    //发送kafka产生告警
                    String sendMsg=createMqMessage(enipAlarmPlatformConfig,enipAlarmPlatformMonitorData,enipNetelement);
                    enipAlarmPlatformMonitorData.setCreateAlarmMsg(sendMsg);
                    sendAlarm(sendMsg);
                }
                enipAlarmPlatformMonitorData.setUpdateTime(dateTime);
                enipAlarmPlatformMonitorMapper.updateEnipAlarmPlatformMonitor(enipAlarmPlatformMonitorData);
                log.info("dealNorthErrorCodeAlarm list2 update:"+enipAlarmPlatformMonitorData.toString());
            }
        }else{
            //该网元该告警类型数据库中不存在
            //与配置表的appear_threshold，做比较判断是否为告警数据
            //判断规则：排除掉数据中other_condition1中的正常错误码，异常错误码数量大于appear_threshold
            //如果是告警数据，执行以下操作
            //将数据时间间隔静态变量(dataTime)与配置表appear_minute告警门限作比较，
            //如果>=配置表appear_minute，向告警平台发送告警信息并存入数据库 status=1，否则将数据存入数据表中 status=2
            enipAlarmPlatformMonitorData=new EnipAlarmPlatformMonitor();
            long tempTotalCount=0;
            for(EdaAlarm edaAlarm:alarmValue){
                if(!set.contains(edaAlarm.getCode())){
                    tempTotalCount=tempTotalCount+Long.parseLong(edaAlarm.getTotalCount());
                }
            }
            if(tempTotalCount>Double.parseDouble(enipAlarmPlatformConfig.getAppearThreshold())){
                enipAlarmPlatformMonitorData.setConfigId(enipAlarmPlatformConfig.getId());
                enipAlarmPlatformMonitorData.setNetId(enipNetelement.getId());
                enipAlarmPlatformMonitorData.setAlarmId("EDA_"+Jdk8DateUtils.now().formatDateTime("yyyyMMddHHmmssSSS"));
                enipAlarmPlatformMonitorData.setCreateTime(dateTime);
                enipAlarmPlatformMonitorData.setCleanMinute(0l);
                long timeTemp=dataTime;
                enipAlarmPlatformMonitorData.setAppearMinute(timeTemp);
                if(timeTemp>=enipAlarmPlatformConfig.getAppearMinute()){
                    enipAlarmPlatformMonitorData.setStatus(1);
                    //发送kafka产生告警
                    String sendMsg=createMqMessage(enipAlarmPlatformConfig,enipAlarmPlatformMonitorData,enipNetelement);
                    enipAlarmPlatformMonitorData.setCreateAlarmMsg(sendMsg);
                    sendAlarm(sendMsg);
                }else{
                    enipAlarmPlatformMonitorData.setStatus(2);
                }
                enipAlarmPlatformMonitorMapper.insertEnipAlarmPlatformMonitor(enipAlarmPlatformMonitorData);
                log.info("dealNorthErrorCodeAlarm insert:"+enipAlarmPlatformMonitorData.toString());
            }
        }
    }



    /**
     * 解析EDA告警文件
     * @param edaAlarmFile
     * @return
     */
    private Map<String,List<EdaAlarm>> analysisFile(File edaAlarmFile) {
        BufferedReader reader = null;
        Map<String,List<EdaAlarm>> resultMap=new HashMap<>();
        try {
            reader = new BufferedReader(new FileReader(edaAlarmFile));
            String tempStr;

            List<EdaAlarm> northEvent=new ArrayList<>();//event 事件
            List<EdaAlarm> northPl=new ArrayList<>();//pl 板卡
            List<EdaAlarm> northErrorCode=new ArrayList<>();//fCode错误码
            String bound="";//标识南向；北向
            int module=0;//标识南向北向中的模块：0、事件 1、板卡 2、详细
            while ((tempStr = reader.readLine()) != null) {
                if(bound.equals("northbound")&&!tempStr.contains("------")){
                    module = getModule(tempStr, northEvent, northPl,northErrorCode, module);
                }
                if(tempStr.contains("southbound")){
                    break;
                }
                if(tempStr.contains("northbound")&&!tempStr.contains("Failed")){
                    bound="northbound";
                    module=0;
                }
            }
            reader.close();
            resultMap.put("northEvent",northEvent);
            resultMap.put("northPl",northPl);
            resultMap.put("northErrorCode",northErrorCode);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (reader != null) {
                try {
                    reader.close();
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
            }
        }
        return resultMap;
    }

    /**
     * 依次解析文件中的各个部分
     * @param module
     * @return
     */
    private static int getModule(String tempStr, List<EdaAlarm> northEvent,List<EdaAlarm> northpl,List<EdaAlarm> northErrorCode, int module) {
        EdaAlarm edaAlarm;
        if(module==0&&!tempStr.contains("z_total")&&!tempStr.contains("PL-board")){
            //System.out.println("事件："+tempStr);
            String [] result=tempStr.replaceAll("\\s+"," ").split(" ");
            if(result.length>5) {
                edaAlarm = new EdaAlarm();
                edaAlarm.setEventName(result[0]);
                edaAlarm.setTotalCount(result[1]);
                edaAlarm.setRateSuc(result[4].replace("%",""));
                northEvent.add(edaAlarm);
            }
        }
        if(module==1&&!tempStr.contains("z_total")&&!tempStr.contains("PL-board")){
            //System.out.println("板卡："+tempStr);
            String [] result=tempStr.replaceAll("\\s+"," ").split(" ");
            if(result.length==7) {
                edaAlarm = new EdaAlarm();
                edaAlarm.setPlName(result[0]);
                edaAlarm.setPl("PL"+result[0].substring(result[0].lastIndexOf("-")));
                edaAlarm.setTotalCount(result[1]);
                edaAlarm.setRateSuc(result[4].replace("%",""));
                edaAlarm.setTimeTotal(result[6]);
                northpl.add(edaAlarm);
            }
        }
        if(module==2&&tempStr.contains("z_total")&&!tempStr.contains("Fail")){
            String [] result=tempStr.replaceAll("\\s+"," ").split(" ");
            if(result.length>=4) {
                edaAlarm = new EdaAlarm();
                edaAlarm.setEventName(result[2]);
                edaAlarm.setTotalCount(result[1]);
                edaAlarm.setCode(result[3]);
                northErrorCode.add(edaAlarm);
            }
        }
        if(tempStr.contains("PL-board")&&tempStr.contains("F_Code")){
            module=2;
        }else if(tempStr.contains("PL-board")&&tempStr.contains("All")){
            module=1;
        }
        return module;
    }

    /**
     * 下载文件
     * @param localDir
     * @param beginTime
     * @param endTime
     * @param ftpInfo
     * @param netId
     * @return
     */
    File download(String localDir, String beginTime,String endTime, FTPInfoVO ftpInfo,Long netId){
        File filePath = new File(localDir);
        // 删除上一批次下载的文件
        if (filePath.exists()) {
            File[] oldxmlTargetFiles = filePath.listFiles();
            for (File file : oldxmlTargetFiles) {
                if (file.exists() && file.isFile()) {
                    log.info("delete last filename"+file.getName());
                    file.delete();
                }
            }
        }

        SFtpCarrierUtil sftpCarrier = new SFtpCarrierUtil(ftpInfo, log);
        try {
            sftpCarrier.connect();
            Vector v = sftpCarrier.listFiles(ftpInfo.getPath());
            if (v.size() > 0) {
                Iterator it = v.iterator();
                while (it.hasNext()) {
                    ChannelSftp.LsEntry entry = (ChannelSftp.LsEntry) it.next();
                    String filename = entry.getFilename();
                    SftpATTRS attrs = entry.getAttrs();
                    if (!attrs.isDir()) {
                        if (filename.endsWith("_KPI.log") && filename.contains(beginTime)&& filename.contains(endTime)) {
                            File tempFile=sftpCarrier.downloadFileKpi(ftpInfo.getPath(), filename, localDir, filename);
                            log.info("netId"+netId+"download filename"+filename);
                            if(null!=tempFile) {
                                return tempFile;
                            }
                        }
                    }
                }
            }
        } catch (SftpException e) {
            log.error("netId"+netId+"SftpException:"+e.getCause());
        } finally {
            sftpCarrier.disconnect();
        }
        return null;
    }

    /**
     * 组装下载连接数据
     * @param list
     * @param i
     * @return
     */
    private FTPInfoVO getFtpInfoVO(List<EnipNetelement> list, int i) {
        FTPInfoVO fTPInfoVO = new FTPInfoVO();
        fTPInfoVO.setHost(list.get(i).getIp4());
        fTPInfoVO.setPort(21);
        fTPInfoVO.setUserName(list.get(i).getUsername());
        fTPInfoVO.setPassword(list.get(i).getPassword());
        fTPInfoVO.setNetElementID(list.get(i).getId());
        fTPInfoVO.setPath(list.get(i).getDownload_path());
        return fTPInfoVO;
    }

    /**
     * 删除数据表中的过期文件
     * @param now
     */
    public void deleteExpireData(Date now) {
        EpcDict epcDict = new EpcDict();
        epcDict.setKeyName(ProcConstant.ProcLogDays);
        epcDict = epcDictService.selectEpcDict(epcDict);
        if (null != epcDict && !"".equals(epcDict.getKeyValue())) {
            String deleteTime = DateUtils.getStatisTime(Integer.parseInt(epcDict.getKeyValue()) * 1440 + 30, now, "yyyy-MM-dd HH:mm:00");
            log.info("deleteTime-------------------------------:" + deleteTime);
            enipAlarmPlatformMonitorMapper.deleteEnipAlarmPlatformMonitorByCreateTime(deleteTime);
        }
    }

    /**
     * 组合发送Kafka的数据
     */
    public String createMqMessage(EnipAlarmPlatformConfig enipAlarmPlatformConfig,EnipAlarmPlatformMonitor enipAlarmPlatformMonitorData,EnipNetelement enipNetelement) {
        String alarmSeq = Jdk8DateUtils.now().formatDateTime("yyyyMMddHHmmssSSS");
        OssAlarmMessage message = new OssAlarmMessage();
        message.setSystemId(enipAlarmPlatformConfig.getSystemid()); //三方系统ID
        message.setMessageType(enipAlarmPlatformConfig.getMessagetype()); //消息分类-一级分类
        message.setSubMessageType("11000"); //消息分类-二级分类
        message.setAlarmTitle(enipAlarmPlatformConfig.getAlarmtitle()); //告警标题
        message.setAlarmStatus(enipAlarmPlatformMonitorData.getStatus().toString()); //告警状态 1：活动告警；0：清除告警
        message.setAlarmId(enipAlarmPlatformMonitorData.getAlarmId()); //告警ID  产生告警与消除告警ID相同
        message.setAlarmSeq(enipAlarmPlatformConfig.getAlarmBusinessType()+"00"+alarmSeq);//告警上报的消息序号 ,每产生一条告警包括消除告警，会产生唯一序号
        message.setAlarmType(enipAlarmPlatformConfig.getAlarmtype()); //告警类型
        message.setOrigSeverity(enipAlarmPlatformConfig.getOrigseverity()); //原始级别
        message.setEventTime(enipAlarmPlatformMonitorData.getUpdateTime());  //告警发生时间
        message.setSpecificProblemID(enipAlarmPlatformConfig.getSpecificproblemid()); //告警问题原因ID。厂家告警ID
        message.setSpecificProblem(enipAlarmPlatformConfig.getSpecificproblem()); //告警问题原因
        message.setObjectType(enipAlarmPlatformConfig.getObjecttype()); //设备类型
        message.setObjectUID(enipNetelement.getId().toString()); //告警定位对象UID
        message.setObjectName(enipNetelement.getNbiAlias());  //网元名
        message.setSubObjectUID(enipNetelement.getId().toString()); //告警定位子对象UID
        message.setSubObjectType(enipAlarmPlatformConfig.getObjecttype());  //告警对象类型
        message.setSubObjectName(enipNetelement.getName()); //告警定位子对象名称

        message.setVendorName(enipAlarmPlatformConfig.getVendorname()); //厂家
        message.setProvince(enipAlarmPlatformConfig.getProvince()); //省份
        message.setCity(enipAlarmPlatformConfig.getCity()); //地市
        message.setAddInfo("");//告警定位信息 为空可以显示告警正文
        return JsonMapper.toJsonString(message);
    }

    private void sendAlarm(String alarmMessage) {
        log.info("send kafka start===================================="+alarmMessage);
        kafkaTemplate.send("ericsson_expert_system", alarmMessage);
        log.info("send kafka end======================================");
    }

}
